iT邦幫忙

2026 iThome 鐵人賽

DAY 10
0
Build on Google AI

30 天用 Google ADK 打造你的「全自動 AI 虛擬團隊」系列 第 10

Day 10 | 網路煉金術:實作非同步爬蟲,抓取最新 AI 趨勢

  • 分享至 

  • xImage
  •  

大家好!歡迎來到「Build on Google AI」工程挑戰的第十天。
在昨天的實戰中,我們成功賦予了代理人 (Agent) 雙手,讓它能透過工具調用 (Tool Calling) 讀取內部專案數據。但在「研發與工程創新」的真實戰場上,資訊往往是動態且分散的。如果我們希望代理人能幫團隊統整最新的 AI 技術趨勢或競品動態,它就必須具備主動向外探索的能力。

今天,我們將結合 Python 的非同步 (Asynchronous) 特性與 Google ADK 的 Function Tool,為代理人打造專屬的「網路爬蟲」,讓它化身為最強大的趨勢煉金術士!

第一步:回顧 ADK 的代理人與工具架構
在開始寫爬蟲之前,我們先快速複習一下 ADK 的核心架構。

  • 在 ADK Python 專案的 agent.py 檔案中,包含了一個 root_agent 的定義,這也是 ADK 代理人中唯一不可或缺的元素。
  • 為了讓代理人具備擴充能力,你可以為代理人定義工具供其使用。
  • 我們只需要更新生成的 agent.py 程式碼,將自定義的工具函式加入代理人的配置中即可。

第二步:實作非同步爬蟲工具
在工程實踐中,網路請求往往是系統中最耗時的瓶頸。為了不讓代理人因為等待網頁回應而卡死,我們將使用 Python 的 aiohttp 來實作一個輕量級的非同步爬蟲工具。

請在 agent.py 中加入以下程式碼:

import re
import aiohttp
from bs4 import BeautifulSoup
from google.adk.agents.llm_agent import Agent


# 常見主題與 Google Cloud Blog 路徑對照表
TOPIC_PATH_MAP = {
    "ai": "products/ai-machine-learning",
    "ml": "products/ai-machine-learning",
    "machine learning": "products/ai-machine-learning",
    "gemini": "products/ai-machine-learning",
    "llm": "products/ai-machine-learning",
    "agent": "products/ai-machine-learning",
    "kubernetes": "products/containers-kubernetes",
    "k8s": "products/containers-kubernetes",
    "container": "products/containers-kubernetes",
    "containers": "products/containers-kubernetes",
    "gke": "products/containers-kubernetes",
    "data": "products/data-analytics",
    "analytics": "products/data-analytics",
    "bigquery": "products/data-analytics",
    "security": "products/identity-security",
    "identity": "products/identity-security",
    "database": "products/databases",
    "databases": "products/databases",
    "networking": "products/networking",
    "network": "products/networking",
    "devops": "products/application-development",
    "app dev": "products/application-development",
    "compute": "products/compute",
    "storage": "products/storage",
    "finops": "products/ai-machine-learning",
    "infrastructure": "topics/ai-infrastructure",
}


# 1. 定義非同步爬蟲工具:爬取 Google Cloud 官方部落格 (https://cloud.google.com/blog/)
async def fetch_cloud_blog_articles(topic: str = "", limit: int = 3) -> dict:
    """
    爬取 Google Cloud 官方部落格 (https://cloud.google.com/blog/) 的最新文章。

    Args:
        topic: 欲查詢的主題或技術關鍵字(例如 'AI', 'Kubernetes', 'BigQuery', 'Security', 'FinOps', 'Networking' 等),若留空則爬取首頁最新文章。
        limit: 回傳文章數量上限(預設 3 篇,範圍 1~10 篇)。

    Returns:
        包含指定主題文章清單、標題、分類、網址與內容摘要的字典。
    """
    limit = max(1, min(limit, 10))
    headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    }

    topic_clean = topic.strip().lower()
    target_path = TOPIC_PATH_MAP.get(topic_clean, "")
    
    # 若有直接對應的主題頁面優先爬取該主題頁,否則爬取 Blog 首頁進行關鍵字過濾
    url = f"https://cloud.google.com/blog/{target_path}" if target_path else "https://cloud.google.com/blog/"

    try:
        async with aiohttp.ClientSession(headers=headers) as session:
            async with session.get(url, timeout=10) as resp:
                if resp.status != 200:
                    return {
                        "status": "error",
                        "message": f"無法連線至 Google Cloud 部落格 (HTTP {resp.status})",
                    }
                html = await resp.text()
                soup = BeautifulSoup(html, "html.parser")

                articles = {}
                for a in soup.find_all("a", href=True):
                    href = a["href"]
                    # 匹配 /blog/products/... 或 /blog/topics/... 結尾的文章連結
                    m = re.search(r"/blog/(products|topics)/([^/]+)/([^/#\?]+)$", href)
                    if m:
                        slug = m.group(3)
                        if slug in ["rss", "latest", "all"]:
                            continue

                        category = m.group(2).replace("-", " ").title()
                        title = a.get("track-name") or a.get_text(strip=True)
                        raw_text = a.get_text(separator=" ", strip=True)

                        # 過濾標題長度合理者
                        if len(title) > 10 and href not in articles:
                            full_url = href if href.startswith("http") else f"https://cloud.google.com{href}"
                            
                            # 若使用者輸入自訂關鍵字且未命中特定路徑,進行字串比對過濾
                            if topic_clean and not target_path:
                                match_corpus = f"{title} {category} {raw_text}".lower()
                                if topic_clean not in match_corpus:
                                    continue

                            articles[href] = {
                                "title": title,
                                "category": category,
                                "url": full_url,
                                "snippet": raw_text[:200],
                            }

                        if len(articles) >= limit:
                            break

                # 若透過特定路徑找不到足夠文章,且非首頁時,可回退搜尋首頁
                if len(articles) == 0 and url != "https://cloud.google.com/blog/":
                    return await fetch_cloud_blog_articles(topic="", limit=limit)

                return {
                    "status": "success",
                    "source": url,
                    "queried_topic": topic or "Latest Overview",
                    "total_found": len(articles),
                    "articles": list(articles.values())[:limit],
                }
    except Exception as e:
        return {"status": "error", "message": f"爬取文章列表異常: {str(e)}"}


async def fetch_cloud_article_detail(url: str) -> dict:
    """
    深度爬取 Google Cloud 官方部落格特定文章的完整段落內容,以供更精準的英翻中與洞察提煉。

    Args:
        url: Google Cloud 文章的完整 URL。

    Returns:
        包含文章標題與核心內文段落的字典。
    """
    if not url.startswith("https://cloud.google.com/blog/"):
        return {"status": "error", "message": "僅支援爬取 https://cloud.google.com/blog/ 網域下的文章"}

    headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    }

    try:
        async with aiohttp.ClientSession(headers=headers) as session:
            async with session.get(url, timeout=10) as resp:
                if resp.status != 200:
                    return {"status": "error", "message": f"文章載入失敗 (HTTP {resp.status})"}
                html = await resp.text()
                soup = BeautifulSoup(html, "html.parser")

                h1 = soup.find("h1")
                title = h1.get_text(strip=True) if h1 else "無標題"
                
                # 萃取內文重要段落(過濾短標籤與導航文字)
                paragraphs = [
                    p.get_text(strip=True) for p in soup.find_all("p")
                    if len(p.get_text(strip=True)) > 35
                ]

                return {
                    "status": "success",
                    "url": url,
                    "title": title,
                    "content": "\n\n".join(paragraphs[:8]),  # 回傳前 8 個核心段落
                }
    except Exception as e:
        return {"status": "error", "message": f"爬取文章內文異常: {str(e)}"}

第三步:將爬蟲掛載至代理人
爬蟲準備就緒後,我們只需要將這個工具配置給代理人。

# 2. 定義系統提示詞與代理人
cloud_instruction = """
你是頂尖的 Google Cloud 技術情報專家與架構研究員。
你的任務是使用 'fetch_cloud_blog_articles' 工具爬取 Google Cloud 官方部落格 (https://cloud.google.com/blog/) 的最新技術發布。
使用者可以自由指定任何主題(例如 AI, Kubernetes, BigQuery, Security, FinOps, DevOps, Networking 等)或篇數。

【執行與英翻中規範】:
1. 收到主題後,立即調用 'fetch_cloud_blog_articles' 抓取最吻合該主題的文章。
2. 必要時可調用 'fetch_cloud_article_detail' 獲取更詳盡的內文。
3. 將英文內容轉化為【繁體中文結構化摘要報告】,每篇文章必須包含:
   - 🎯【繁體中文標題】(保留具代表性的英文專有名詞如 GKE, Vertex AI, BigQuery)
   - 🏷️【分類與原文網址】(附上完整的原文 Link)
   - 💡【3~5 個核心重點精華】(條列式繁中重點,清楚說明解決什麼問題、有何新功能)
   - 🚀【技術洞察與架構建議】(為工程團隊提出具體實務建議)
"""

root_agent = Agent(
    model="gemini-flash-latest",
    name="google_cloud_blog_researcher",
    description="爬取 Google Cloud 官方部落格 (https://cloud.google.com/blog/),支援自由選擇主題並自動進行英翻中深度結構化分析。",
    instruction=cloud_instruction,
    tools=[fetch_cloud_blog_articles, fetch_cloud_article_detail],
)

第四步:工程創新視角反思
這不僅僅是一個爬蟲,這是一次「系統解耦」的完美示範。
傳統的自動化腳本一旦遇到網頁改版,整個流程就會崩潰。但在 ADK 的 Agent 架構下,爬蟲工具只負責提供「原始的 DOM 數據」或「粗略的文字陣列」,而資料清洗、語意理解與摘要萃取的重責大任,則交由強大的 Gemini 模型來處理。這種高容錯率的設計,大幅提升了系統在真實場景中的穩定運作與可維護性。

https://ithelp.ithome.com.tw/upload/images/20260911/201216437lvlUG3Hsj.png

小結
今天,我們透過非同步爬蟲與 ADK Tool Calling 的結合,成功將代理人的感知能力延伸到了廣闊的網際網路。它現在不僅能讀取內部資料,還能主動挖掘外部趨勢。


上一篇
Day 09 | 賦予雙手:Google ADK Tool Calling 基礎實作
下一篇
Day 11 | 髒數據清洗:利用 Gemini 處理非結構化情報
系列文
30 天用 Google ADK 打造你的「全自動 AI 虛擬團隊」20
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言